package test.study.bigdata.flink.state.side_output;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

@Slf4j
public class SideOutputJob {

    @SneakyThrows
    public static void main(String[] args) {
        //
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.fromElements("abc", "b123", "bbb", "aaa");

        // 定义旁路
        OutputTag<String> tagA = new OutputTag<String>("a") {
        };
        OutputTag<String> tagB = new OutputTag<String>("b") {
        };
        //
        SingleOutputStreamOperator<String> ds = source.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> collector) throws Exception {
                //使用旁路
                if (value.startsWith("a")) {
                    ctx.output(tagA, value);
                } else if (value.startsWith("b")) {
                    ctx.output(tagB, value);
                }
                //collector.collect(value);
            }
        });
        //获取旁路，并添加 sink
        ds.getSideOutput(tagA)
                .addSink(new PrintSinkFunction<>())
                .name("testa");
        ds.getSideOutput(tagB)
                .addSink(new PrintSinkFunction<>())
                .name("testb");
        ds.print(">>>>>> ");

        //
        env.execute("Flink-Checkpoint-Job");
    }
}
